# coding: utf-8
import MySQLdb
import pandas as pd
import pytz
from datetime import datetime
import pymongo
from time import time
import os
import logging
import re
import timeUtils_4from_mongo as timeUtils

"""
把历史行情数据导入到mongodb
"""


def _get_stocks_from_db():
    # connect to mysql
    conn = MySQLdb.connect(host='rdsqtehrv8tqh7v60yvujpublic.mysql.rds.aliyuncs.com',
                           user='tradingfloor',
                           passwd='tradingfloor123',
                           db='tradingreason',
                           charset='utf8')
    # conn = MySQLdb.connect(host='localhost',
    #                   user='root',
    #                   passwd='root',
    #                   db='tradingreasontt',
    #                   charset='utf8')
    cur = conn.cursor()
    res = []
    cur.execute("SELECT code FROM future_info WHERE code like 'al%'")
    for (symbol,) in cur:
        res.append(symbol)
    return res

logger = logging.basicConfig(level=logging.INFO)
t0 = time()
tz = pytz.timezone(pytz.country_timezones('cn')[0])
start = datetime(2010, 1, 1, 8, 0, tzinfo=tz)
end = datetime(2015, 1, 1, 8, 0, tzinfo=tz)

log = open('resample_log.txt', 'a')
td = datetime.today()
log.write('#####################################################\n')
log.write('task begins at: ' + td.strftime('%Y-%m-%d %H:%M:%S') + '\n')
log.write('data range:\n')
log.write('start date :' + start.strftime('%Y-%m-%d %H:%M:%S') + '\n')
log.write('end date :' + end.strftime('%Y-%m-%d %H:%M:%S') + '\n')

client = pymongo.MongoClient('121.40.212.219', 27017)  # '121.40.212.219', 27017
db = client.future_data

# symbols = _get_stocks_from_db()
symbols = list(db.day.distinct("symbol"))
for symbol in symbols:
    symbol = symbol.encode('UTF-8')
    symbolUP = symbol.upper()
    if symbolUP.startswith("AL1") or symbolUP.startswith("AL7"):                              # 需要改动的
        num = re.findall(r"\d+\.?\d*", symbol)
        if num[0] < '1601' or num[0] == '77777':
            logging.info("------- little -------")
        else:
            logging.info("------- bigger -------")
            continue
    else:
        continue
    logging.info("------- "+symbol+" -------")
    cur = db.one_min_data.find({'symbol': symbol, 'dateTime': {'$gte': start, '$lte': end}}, {'_id': 0}) \
            .sort("dateTime", pymongo.ASCENDING)
    df = pd.DataFrame(list(cur))

    if len(df.index) > 1:

        df = df[df['close'] != 0]
        df = df[df['low'] > 5]
        tz = pytz.timezone(pytz.country_timezones('cn')[0])
        df['dateTime'] = df['dateTime'].apply(lambda x: tz.localize(x))
        df['symbol'] = symbol
        # if len(df.index) > 1:
        #     db.one_min_data.insert_many(df.reset_index().to_dict(orient='records'))
        # else:
        #     logging.info("-------one min done-------  " + symbol + "--- is null ---")
        # logging.info("-------one min done-------")
        df_m30 = df.copy()
        df_m60 = df.copy()
        df_m120 = df.copy()
        df_day = df.copy()
        df = df.set_index('dateTime')

        # ---------------write in 5 minute data----------------
        df_m5 = pd.DataFrame([])
        df_m5['open'] = df['open'].resample('5min', how='first', closed='right', label='left').dropna()
        df_m5['close'] = df['close'].resample('5min', how='last', closed='right', label='left').dropna()
        df_m5['high'] = df['high'].resample('5min', how='max', closed='right', label='left').dropna()
        df_m5['low'] = df['low'].resample('5min', how='min', closed='right', label='left').dropna()
        # df_m5['dateTime'] = df['dateTime'].resample('5min', how='last', closed='right', label='right').dropna()
        df_m5['volume'] = df['volume'].resample('5min', how='sum', closed='right', label='left').dropna()
        df_m5['symbol'] = symbol
        # if len(df_m5.index) > 1:
        #     db.min_5_data.insert_many(df_m5.reset_index().to_dict(orient='records'))
        # else:
        #     logging.info("-------5 min done-------  " + symbol + "--- is null ---")
        # logging.info("-------5 min done-------")
        # ---------------write in 15 minute data----------------
        df_m15 = pd.DataFrame([])
        df_m15['open'] = df_m5['open'].resample('15min', how='first', closed='right', label='right').dropna()
        df_m15['close'] = df_m5['close'].resample('15min', how='last', closed='right', label='right').dropna()
        df_m15['high'] = df_m5['high'].resample('15min', how='max', closed='right', label='right').dropna()
        df_m15['low'] = df_m5['low'].resample('15min', how='min', closed='right', label='right').dropna()
        df_m15['volume'] = df_m5['volume'].resample('15min', how='sum', closed='right', label='right').dropna()
        df_m15['symbol'] = symbol
        # if len(df_m15.index) > 1:
        #     db.min_15_data.insert_many(df_m15.reset_index().to_dict(orient='records'))
        # else:
        #     logging.info("-------15 min done-------  " + symbol + "--- is null ---")
        # logging.info("-------15 min done-------")
        # ---------------write in 30 minute data----------------
        df_m30['dateTime'] = timeUtils.Minute30New(df_m30['dateTime'], symbol)
        df_m30['symbol'] = symbol
        grouped30 = df_m30.groupby('dateTime')
        functions = {'open': 'first', 'close': 'last', 'high': 'max', 'low': 'min', 'volume': 'sum'}
        rsDF30 = grouped30.agg(functions)
        rsDF30['symbol'] = symbol
        # if len(rsDF30.index) > 1:
        #     db.min_30_data.insert_many(rsDF30.reset_index().to_dict(orient='records'))
        # else:
        #     logging.info("-------30 min done-------  " + symbol + "--- is null ---")
        # logging.info("-------30 min done-------")
        # ---------------write in 60 minute data----------------
        df_m60['dateTime'] = timeUtils.Minute60New(df_m60['dateTime'], symbol)
        df_m60['symbol'] = symbol
        grouped = df_m60.groupby('dateTime')
        functions = {'open': 'first', 'close': 'last', 'high': 'max', 'low': 'min', 'volume': 'sum'}
        rsDF = grouped.agg(functions)
        rsDF['symbol'] = symbol
        if len(rsDF.index) > 1:
            db.min_60_data.insert_many(rsDF.reset_index().to_dict(orient='records'))
        else:
            logging.info("-------60 min done-------  " + symbol + "--- is null ---")
        logging.info("-------60 min done-------")
        # ---------------write in 120 minute data----------------
        df_m120['dateTime'] = timeUtils.Minute120New(df_m120['dateTime'], symbol)
        df_m120['symbol'] = symbol
        grouped120 = df_m120.groupby('dateTime')
        functions = {'open': 'first', 'close': 'last', 'high': 'max', 'low': 'min', 'volume': 'sum'}
        rsDF120 = grouped120.agg(functions)
        rsDF120['symbol'] = symbol
        # if len(rsDF120.index) > 1:
        #     db.min_120_data.insert_many(rsDF120.reset_index().to_dict(orient='records'))
        # else:
        #     logging.info("-------120 min done-------  " + symbol + "--- is null ---")
        # logging.info("-------120 min done-------")
        # ---------------write in day data----------------
        df_day['dateTime'] = timeUtils.Day(df_day['dateTime'])
        df_day['symbol'] = symbol
        grouped_day = df_day.groupby('dateTime')
        functions = {'open': 'first', 'close': 'last', 'high': 'max', 'low': 'min', 'volume': 'sum'}
        rsDF_day = grouped_day.agg(functions)
        rsDF_day['symbol'] = symbol
        # if len(rsDF_day.index) >= 1:
        #     db.day.insert_many(rsDF_day.reset_index().to_dict(orient='records'))
        # else:
        #     logging.info("-------day done-------  " + symbol + "--- is null ---")
        # logging.info("-------day done-------")
    else:
        print(symbol + "is null")

# gtm = df['open'].resample('dateTime', how='first')

# if the first row of pChange is NaN, then we ignore the first row
#     d = tz.localize(datetime(2015, 9, 30, 0, 0))
#     if math.isnan(df.pChange.iloc[0]) and df.dateTime.iloc[0] == d:
#         if df[1:].empty:
#             pass
#         else:
#             db.original_prices.insert_many(df[1:].to_dict(orient='records'))
#     else:
#         db.original_prices.insert_many(df.to_dict(orient='records'))
#     fi.write(filename + '\n')
#
# fi.close()
# print('finished in %.2fs' % (time() - t0))
